package com.doit.day01;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.Properties;

public class KafkaProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        //创建一个props对象
        Properties props = new Properties();
        //设置一些参数
//        org.apache.kafka.common.serialization.Serializer   反射
        //org.apache.kafka.common.serialization.StringSerializer
        /**
         * 这三个是必须配置的参数
         */
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,MyConfig.HOST_AND_PORT);
        //选择性配置的参数
//        props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"163840");
//分区器
//        props.setProperty("partitioner.class",)
//        props.setProperty("acks","-1");
        //创建一个生产者的对象
        //我们往kafka里面写数据，都是需要指定 value
        KafkaProducer<String, String> producer = new KafkaProducer<String,String>(props);

        UserInfo userInfo = new UserInfo(1, "zss", 18, "male");

        ArrayList<String> list = new ArrayList<>();
        list.add("male");
        list.add("female");

        while (true){
            userInfo.setId(RandomUtils.nextInt(100,10000));
            userInfo.setAge(RandomUtils.nextInt(20,80));
            userInfo.setName(RandomStringUtils.randomAlphabetic(10));
            userInfo.setGender(list.get(RandomUtils.nextInt(0,2)));
            //将userInfo对象转换成json串
            String jsonString = JSON.toJSONString(userInfo);
            //对象去调用方法发送数据
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test3", jsonString);
            //你居然要一个producerRecord
            producer.send(producerRecord);
            producer.flush();
            Thread.sleep(1000);
        }


        //关闭资源



    }
}
